package org.ykhl.mq.commons.pusher;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.ykhl.mq.commons.BaseMessageKafkaImpl;
import org.ykhl.mq.commons.Constants;
import org.ykhl.mq.commons.Utils;
import org.ykhl.mq.commons.BaseMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Future;

/**
 * Created by zkf on 2015/11/25.
 */
public class MessagePusherKafkaImpl extends MessagePusher {
    private static final Logger LOG = LoggerFactory.getLogger(MessagePusherKafkaImpl.class);
    private Producer<String, String> producer;

    public MessagePusherKafkaImpl(String serverAddress, String authKey,
                                  String securityFilePath, Constants.Role role) {
        super(serverAddress, authKey, securityFilePath, role);

    }

    public MessagePusherKafkaImpl(String serverAddress, String authKey,
                                  String securityFilePath, Constants.Role role,
                                  String topicPrefix, String producerPrefix) {
        super(serverAddress, authKey, securityFilePath, role, topicPrefix, producerPrefix);

    }

    @Override
    protected void doStart() throws Exception {
        producer = new KafkaProducer<>(Utils.genKafkaProducerProps(
                serverAddress, useSsl, securityFilePath));
    }

    @Override
    protected void doStop() throws Exception {
        producer.close();
    }

    @Override
    protected boolean doSendMessage(PushMessage pushMessage) throws Exception {
        Future<RecordMetadata> future = doSendMessageAsync(pushMessage);
        producer.flush();
        future.get();
        return true;
    }

    @Override
    protected Future<RecordMetadata> doSendMessageAsync(PushMessage pushMessage) {
        BaseMessageKafkaImpl msg = trans2Message(pushMessage);
        ProducerRecord<String, String> record = msg.toProducerRecord();

        return producer.send(record);
    }

    @Override
    protected BaseMessageKafkaImpl trans2Message(PushMessage pushMessage){
        BaseMessageKafkaImpl msg = new BaseMessageKafkaImpl(topic, pushMessage.getMsg());

        msg.setAppServerId(authKey);

        msg.setMessageType(BaseMessage.MessageType.PUSH);
        msg.setDataType(pushMessage.getDataType());
        msg.setDataVersion(pushMessage.getDataVersion());
        msg.setTimestamp(Long.toString(System.currentTimeMillis()));
        msg.setFlowId(msg.getFlowId());
        msg.setExtendSet(msg.getExtendSet());

        if (pushMessage instanceof RequestResponse){
            RequestResponse rr = (RequestResponse) pushMessage;
            msg.setRequestId(rr.getRequestId());
            msg.setFid(rr.getFid());
            msg.setDeviceSign(rr.getDeviceSign());

            if (role == Constants.Role.HOSPITAL) {
                msg.setMessageType(BaseMessage.MessageType.RESPONSE);
                msg.setAppServerId(rr.getRequestAppServerId());
            }
            else {
                msg.setMessageType(BaseMessage.MessageType.REQUEST);
            }
            msg.setAppSecretKey(rr.getAppSecretKey());
            msg.setExtendSet(rr.getExtendSet());
        }
        return msg;
    }
}
